package com.eshop.eshopcache.kafka;

import com.alibaba.fastjson.JSONObject;
import com.eshop.eshopcache.enums.ServiceTypeEmun;
import com.eshop.eshopcache.enums.ZkLock;
import com.eshop.eshopcache.model.ProductInfo;
import com.eshop.eshopcache.model.ShopInfo;
import com.eshop.eshopcache.service.CacheService;
import com.eshop.eshopcache.utils.DateUtils;
import com.eshop.eshopcache.zk.ZooKeeperSession;
import lombok.extern.slf4j.Slf4j;

/**
 * @author: Xukai
 * @description:
 * @createDate: 2018/7/26 14:01
 * @modified By:
 */
@Slf4j
public class KafkaMessageProcessor implements Runnable {

    private String message;

    private CacheService cacheService;

    public KafkaMessageProcessor(String message, CacheService cacheService) {
        this.message = message;
        this.cacheService = cacheService;
    }

    @Override
    public void run() {
        // 将message转换成json对象
        JSONObject messageJSONObject = JSONObject.parseObject(message);
        // 提取消息对应的服务的标识
        String serviceId = messageJSONObject.getString("serviceId");
        // 如果是商品信息服务
        if (ServiceTypeEmun.PRODUCT_INFO_SERVICE.getCode().equals(serviceId)) {
            processProductInfoChangeMessage(messageJSONObject);
        } else if (ServiceTypeEmun.SHOP_INFO_SERVICE.getCode().equals(serviceId)) {
            processShopInfoChangeMessage(messageJSONObject);
        }
    }

    /**
     * 处理商品信息变更的消息
     *
     * @param messageJSONObject
     */
    private void processProductInfoChangeMessage(JSONObject messageJSONObject) {
        // 提取商品id
        Long productId = messageJSONObject.getLong("productId");
        // 调用商品信息服务的接口
        // 这里直接模拟返回结果
        String productInfoJSON = "{\"id\": " + productId + ", \"name\": \"iphone7手机\", \"price\": 5599, \"pictureList\":\"a.jpg,b.jpg\", \"specification\": \"iphone7的规格\", \"service\": \"iphone7的售后服务\", \"color\": \"红色,白色,黑色\", \"size\": \"5.5\", \"shopId\": 1, \"modifiedTime\": \"2017-01-01 12:00:00\"}";
        ProductInfo productInfo = JSONObject.parseObject(productInfoJSON, ProductInfo.class);
        // 在将数据直接写入本地缓存+redis缓存之前，应该先获取一个zk的分布式锁
        ZooKeeperSession zooKeeperSession = ZooKeeperSession.getInstance();
        zooKeeperSession.acquireDistributedLock(ZkLock.PRODUCT_LOCK.getType(), productId);
        // 获取到了锁
        // 先从本地缓存中获取数据
        ProductInfo productInfoFromLocalCache = cacheService.getProductInfoFromLocalCache(productId);
        // 判断本地缓存中的数据是否为空，同时该数据的更新时间是否早于新数据的时间
        if (productInfoFromLocalCache == null || DateUtils.after(productInfo.getModifiedTime(), productInfoFromLocalCache.getModifiedTime())) {
            // 写入本地缓存
            cacheService.saveProductInfo2LocalCache(productInfo);
        }
        // 先从redis中获取数据
        ProductInfo productInfoFromRedisCache = cacheService.getProductInfoFromRedisCache(productId);
        // 判断redis中的数据是否为空，同时该数据的更新时间是否早于新数据的时间
        if (productInfoFromRedisCache == null || DateUtils.after(productInfo.getModifiedTime(), productInfoFromRedisCache.getModifiedTime())) {
            cacheService.saveProductInfo2RedisCache(productInfo);
        }
        // 释放分布式锁
        zooKeeperSession.releaseDistributedLock(ZkLock.PRODUCT_LOCK.getType(), productId);
    }

    /**
     * 处理店铺信息变更的消息
     *
     * @param messageJSONObject
     */
    private void processShopInfoChangeMessage(JSONObject messageJSONObject) {
        // 提取商品id
        Long shopId = messageJSONObject.getLong("shopId");
        // 调用商品店铺服务的接口
        // 这里直接模拟返回结果
        String shopInfoJSON = "{\"id\": " + shopId + ", \"name\": \"小王的手机店\", \"level\": 5, \"goodCommentRate\":0.99, \"modifiedTime\": \"2017-01-01 12:00:00\"}";
        ShopInfo shopInfo = JSONObject.parseObject(shopInfoJSON, ShopInfo.class);
        // 在将数据直接写入本地缓存+redis缓存之前，应该先获取一个zk的分布式锁
        ZooKeeperSession zooKeeperSession = ZooKeeperSession.getInstance();
        zooKeeperSession.acquireDistributedLock(ZkLock.SHOP_LOCK.getType(), shopId);
        // 获取到了锁
        // 先从本地缓存中获取数据
        ShopInfo shopInfoFromLocalCache = cacheService.getShopInfoFromLocalCache(shopId);
        // 判断本地缓存中的数据是否为空，同时该数据的更新时间是否早于新数据的时间
        if (shopInfoFromLocalCache == null || DateUtils.after(shopInfo.getModifiedTime(), shopInfoFromLocalCache.getModifiedTime())) {
            // 写入本地缓存
            cacheService.saveShopInfo2LocalCache(shopInfo);
        }
        // 先从redis中获取数据
        ShopInfo shopInfoFromRedisCache = cacheService.getShopInfoFromRedisCache(shopId);
        // 判断redis中的数据是否为空，同时该数据的更新时间是否早于新数据的时间
        if (shopInfoFromRedisCache == null || DateUtils.after(shopInfo.getModifiedTime(), shopInfoFromRedisCache.getModifiedTime())) {
            // 写入redis缓存
            cacheService.saveShopInfo2RedisCache(shopInfo);
        }
        // 释放分布式锁
        zooKeeperSession.releaseDistributedLock(ZkLock.SHOP_LOCK.getType(), shopId);
    }
}
